package com.zhangh.summer.admin.support.ds.rocketmq;
import com.zhangh.summer.admin.support.spring.util.SpringBeanUtil;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.springframework.core.env.Environment;

/**
 * @description: TODO
 * @author: zhanghui  zhanghlucky@163.com
 * @create: 2021-07-03 23:07
 **/
public abstract class BaseConsumer {
    private Environment env;
    protected  String nameServer;
    protected abstract MessageListenerConcurrently concurrently();

    public void receiveMessage(String group, String topic,MessageListenerConcurrently concurrently) throws Exception{
        env= SpringBeanUtil.getBeanByType(Environment.class);
        nameServer=env.getProperty("ds.rocketmq.nameServer");

        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(group);
        consumer.setNamesrvAddr(nameServer);
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        consumer.subscribe(topic, "*");
        consumer.registerMessageListener(concurrently);
        consumer.start();

    }
}
